package com.atguigu.dga.governance.assess.calc;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.ds.bean.TDsTaskInstance;
import com.atguigu.dga.governance.assess.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.meta.constant.CodeConst;
import com.atguigu.dga.util.HttpUtil;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

@Component("DATA_SKEW")
public class DataSkewAssessor extends Assessor {

    @Value("${history-server.api.url}")
    String historyServerApiUrl;

    //检查是否有数据倾斜，如果某个stage的最大任务耗时超过其他任务平均耗时耗时的{percent}%, 只检查耗时超过{stage_dur_seconds}秒的stage。
    //存在倾斜给0分，不存在给10分。

    //1 取得最新的且成功的 taskInstance
    //2  从taskInstance 提取 applink 也就是yarn app id
    //3 http://hadoop102:18080/api/v1/applications/application_1684083580862_0012   整体的任务信息  提取成功的attemptId
    //
    //4  http://hadoop102:18080/api/v1/applications/application_1684083580862_0012/1/stages
    //  获得stage 列表的信息   提取 stageId  过滤掉只有一个task的stage ,过滤掉处理时长特别短的stage
    //
    //
    //5 http://hadoop102:18080/api/v1/applications/application_1684083580862_0012/1/stages/1
    // 获得单个stage的具体信息 包括其下所有task的信息
    //         从task中提取duration   根据 duration  进行计算     -> stage 的  任务数据    最大的任务耗时   其他任务耗时的平均值  =  （所有其他任务的总耗时-最大任务耗时） /（总任务数-1）


    // 最大的任务耗时 - 其他任务耗时的平均值 / 其他任务耗时的平均值   如果 大于 参考占比 则差评

    @Override
    public void checkProblem(GovernanceAssessDetail governanceAssessDetail, AssessParam assessParam) throws Exception {
        //1 取得最新的且成功的 taskInstance
        if(assessParam.getTDsTaskInstanceList()==null){
            return;
        }
        List<TDsTaskInstance> dsTaskInstanceList = assessParam.getTDsTaskInstanceList();
        TDsTaskInstance lastSuccessTaskInstance=null;
        for (TDsTaskInstance tDsTaskInstance : dsTaskInstanceList) {
            if(tDsTaskInstance.getState().intValue()== CodeConst.TASK_STATE_SUCCESS){
               if(lastSuccessTaskInstance==null){
                   lastSuccessTaskInstance=tDsTaskInstance;
               }else  if(lastSuccessTaskInstance.getSubmitTime().compareTo(tDsTaskInstance.getSubmitTime())<0){  //前小后大 ,则替换
                   lastSuccessTaskInstance=tDsTaskInstance;
               }
            }
        }
        if(lastSuccessTaskInstance==null){
            governanceAssessDetail.setAssessComment("未发现成功的任务实例");
            return;
        }
        //2  从taskInstance 提取 applink 也就是yarn app id
        String yarnAppId = lastSuccessTaskInstance.getAppLink();
        //3  提取成功的attemptId
        String attempId=  getAttemptIdByYarnAppId(yarnAppId);
        if(attempId==null){
            return;
        }

        //4  http://hadoop102:18080/api/v1/applications/application_1684083580862_0012/1/stages
        //  获得stage 列表的信息   提取 stageId  过滤掉只有一个task的stage ,过滤掉处理时长特别短的stage
        //
        String metricParamsJson = assessParam.getGovernanceMetric().getMetricParamsJson();
        JSONObject paramJsonObj = JSON.parseObject(metricParamsJson);
        Integer stageDurSeconds = paramJsonObj.getInteger("stage_dur_seconds");

        List<String> stageIdList= getStageIdList(yarnAppId,attempId,stageDurSeconds);


        //5 http://hadoop102:18080/api/v1/applications/application_1684083580862_0012/1/stages/1
        // 获得单个stage的具体信息 包括其下所有task的信息  提取 1  任务数 2最大任务时长 3 总时长
        List<StageInfo>  stageInfoList=  getStageInfoList(  yarnAppId,   attempId,stageIdList);


        //6 考评  最大的任务耗时  总时长
        //    其他任务时长平均值=     (总时长-最大的任务耗时）/任务数-1
        //  （最大的任务耗时 -其他任务时长平均值 ） /  其他任务时长平均值 == 超过时长占比
        //  如果 超过时长占比>参考值 则差评
        Integer paramPercent = paramJsonObj.getInteger("percent");
        List<String> beyondStageCommentList=new ArrayList<>();
        for (StageInfo stageInfo : stageInfoList) {
            Integer otherAvgDuration= (stageInfo.getTaskTotalDurationSec()-stageInfo.getMaxTaskDurationSec()) / (stageInfo.getNumTasks()-1 ) ;
            Integer beyondDurationPercent= (stageInfo.getMaxTaskDurationSec()-otherAvgDuration) *100 /otherAvgDuration;
            if(paramPercent<beyondDurationPercent){
                beyondStageCommentList.add("stageId"+stageInfo.getStageId()+",其他任务平均时长："+otherAvgDuration+",最大任务耗时:"+ stageInfo.getMaxTaskDurationSec());
            }

        }

        //7总评
        if(beyondStageCommentList.size()>0){
            governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
            governanceAssessDetail.setAssessProblem("存在数据倾斜："+ StringUtils.join(beyondStageCommentList,";") );
        }



    }

    private List<StageInfo> getStageInfoList(String yarnAppId, String attempId, List<String> stageIdList) throws Exception {
        List<StageInfo> stageInfoList= new ArrayList<>();

        for (String stageId : stageIdList) {
            StageInfo stageInfo = new StageInfo();
            stageInfo.setStageId(stageId);

            String stageTaskUrl=historyServerApiUrl+"/"+yarnAppId+"/"+attempId+"/stages/"+stageId;
            String stageTaskJson = HttpUtil.get(stageTaskUrl);
            List<JSONObject> stageTaskJsonList = JSON.parseArray(stageTaskJson, JSONObject.class);
            JSONObject  stageTaskJsonObj = stageTaskJsonList.get(0);
            JSONObject tasksJsonObj = stageTaskJsonObj.getJSONObject("tasks");


            for (Object taskObj : tasksJsonObj.values()) {
                JSONObject taskJsonObj = (JSONObject) taskObj;
                Integer duration = taskJsonObj.getInteger("duration");
                stageInfo.setNumTasks(stageInfo.getNumTasks()+1);//任务数
                stageInfo.setTaskTotalDurationSec(stageInfo.getTaskTotalDurationSec()+duration);  //总值
                stageInfo.setMaxTaskDurationSec(Math.max(stageInfo.getMaxTaskDurationSec(),duration)); //最大值
            }
            stageInfoList.add(stageInfo);
        }
        return stageInfoList;
    }

    //根据yarnid 和 attemtpId 获取 有多个task的stage  ，过滤时长较短 依据 stage_dur_seconds
    private List<String> getStageIdList(String yarnAppId, String attempId ,Integer stageDurSeconds) throws Exception {
        String stagesListUrl=historyServerApiUrl+"/"+yarnAppId+"/"+attempId+"/stages";
        String stageListJson = HttpUtil.get(stagesListUrl);
        List<JSONObject> stageJsonObjList = JSON.parseArray(stageListJson, JSONObject.class);
        List<String> stageIdList = stageJsonObjList.stream().filter(stageJsonObj -> {  //过滤 ：1 任务数大于1  2 总stage时长大于参考值
                    Integer numTasks = stageJsonObj.getInteger("numTasks");
                    Integer executorDeserializeTime = stageJsonObj.getInteger("executorDeserializeTime");
                    Integer executorRunTime = stageJsonObj.getInteger("executorRunTime");
                    Integer totalStageTime = executorRunTime + executorDeserializeTime;
                    if (numTasks > 1 && totalStageTime > stageDurSeconds) {
                        return true;
                    }
                    return false;
                }).map(stageJsonObj -> stageJsonObj.getString("stageId"))
                .collect(Collectors.toList());

        return stageIdList;

    }

    //3  提取成功的attemptId
    private String getAttemptIdByYarnAppId(String yarnAppId) throws Exception {

        String attempUrl=historyServerApiUrl+"/"+yarnAppId;
        String attempJson = HttpUtil.get(attempUrl);
        JSONObject jobJsonObj = JSON.parseObject(attempJson);
        JSONArray attemptJSONArray = jobJsonObj.getJSONArray("attempts");

        for (int i = 0; i < attemptJSONArray.size(); i++) {   //取成功的最大attemptId
            JSONObject attemptJsonObj = attemptJSONArray.getJSONObject(i);
            if(attemptJsonObj.getBoolean("completed")){  //成功的尝试有且只会有一个
                return  attemptJsonObj.getString("attemptId");
            }
        }
        return null;
    }

    @Data
    private class StageInfo {
        String stageId;
        Integer maxTaskDurationSec=0;

        Integer taskTotalDurationSec=0;

        Integer numTasks=0;

    }
}
